package ai.fabu.dataplatform.service.impl;


import ai.fabu.dataplatform.bean.bagdata.BagData;
import ai.fabu.dataplatform.bean.framedata.FrameData;
import ai.fabu.dataplatform.bean.framedata.GetBinData;
import ai.fabu.dataplatform.bean.groupdata.GroupData;
import ai.fabu.dataplatform.bean.metadata.GetMetadata;
import ai.fabu.dataplatform.bean.task.GroupDataTask;
import ai.fabu.dataplatform.bean.topicdata.TopicData;
import ai.fabu.dataplatform.bean.util.IndexData;
import ai.fabu.dataplatform.bean.util.Pair;
import ai.fabu.dataplatform.constant.ResultCode;
import ai.fabu.dataplatform.help.Result;
import ai.fabu.dataplatform.service.ApiService;
import ai.fabu.dataplatform.utils.FileUtil;
import ai.fabu.dataplatform.utils.ThreadManager;
import ai.fabu.dataplatform.utils.TimeUtil;
import com.mongodb.MongoException;
import com.mongodb.client.result.UpdateResult;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.UncategorizedMongoDbException;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.validation.constraints.NotNull;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.Callable;

import static ai.fabu.dataplatform.constant.ConstantName.*;
import static ai.fabu.dataplatform.constant.ConstantValue.*;
import static ai.fabu.dataplatform.utils.FileUtil.readByteFile;


@Slf4j
@Service
public class ApiServiceImpl implements ApiService {
    @Autowired
    MongoOperations mongoOperations;
    private final static Set<String> groupPathKey = new HashSet<String>(){
        {
            add(LOCALIZATION_IMAGE_PATH);
        }
    };
    private final static Set<String> bagPathKey = new HashSet<String>(){
        {

        }
    };
    private final static Set<String> topicPathKey = new HashSet<String>(){
        {

        }
    };
    private final static Set<String> framePathKey = new HashSet<String>(){
        {
            add(PATH);
        }
    };


    private List<Pair<String, byte[]>> traversalMetadata(Map<String, byte[]> metadata, final Set<String> keySet) throws IOException {
        List<Pair<String, byte[]>> result = new ArrayList<>();
        for (Map.Entry<String, byte[]> entry : metadata.entrySet()){
            String key = entry.getKey();
            byte[] value;
            if(keySet.contains(key)){
                value = FileUtil.readByteFile(new String(entry.getValue()));
            }
            else{
                value = entry.getValue();
            }
            result.add(new Pair<>(key, value));
        }
        return result;
    }

    private Query getDifferentQuery(String groupName, String bagName, String topicName, int frameNo, int type){
        Query query = new Query();
        switch (type){
            case GROUP_TYPE:
                if(!groupName.isEmpty())  query.addCriteria(Criteria.where(GROUP_NAME).is(groupName));
                break;
            case BAG_TYPE:
                if(!groupName.isEmpty()){
                    query.addCriteria(Criteria.where(GROUP_NAME).is(groupName));
                    if(!bagName.isEmpty())  query.addCriteria(Criteria.where(BAG_NAME).is(bagName));
                }
                break;
            case TOPIC_TYPE:
                if(!groupName.isEmpty()){
                    query.addCriteria(Criteria.where(GROUP_NAME).is(groupName));
                    if(!bagName.isEmpty()){
                        query.addCriteria(Criteria.where(BAG_NAME).is(bagName));
                        if(!topicName.isEmpty())  query.addCriteria(Criteria.where(TOPIC_NAME).is(topicName));
                    }
                }
                break;
            case FRAME_TYPE:
                if(!groupName.isEmpty()){
                    query.addCriteria(Criteria.where(GROUP_NAME).is(groupName));
                    if(!bagName.isEmpty()){
                        query.addCriteria(Criteria.where(BAG_NAME).is(bagName));
                        if(!topicName.isEmpty()){
                            query.addCriteria(Criteria.where(TOPIC_NAME).is(topicName));
                            if(frameNo > 0)  query.addCriteria(Criteria.where(FRAME_NO).is(frameNo));
                        }
                    }
                }
                break;
            default:
                log.error("getDifferentQuery: type error");
        }
        return query;
    }


    private Result<Query> getFilterQuery(String groupName, String bagName, String topicName, int frameNo, int type){
        Result<Query> result = new Result<>();
        Query query = new Query();
        if(groupName.isEmpty()){
            Query taskQuery = new Query();
            taskQuery.addCriteria(Criteria.where(COMPLETE_STATUS).is(PROCESSED_STATUS));
            taskQuery.fields().include(GROUP_NAME);
            List<GroupDataTask> groupDataTaskList = mongoOperations.find(taskQuery, GroupDataTask.class);
            List<String> groupNameList = new ArrayList<>();
            groupDataTaskList.forEach(groupDataTask -> groupNameList.add(groupDataTask.getGroupName()));
            query.addCriteria(Criteria.where(GROUP_NAME).in(groupNameList));

            //query.addCriteria(Criteria.where(GROUP_NAME).nin(TaskStatusTable.taskList));
        }
        else{
            Query taskQuery = new Query();
            taskQuery.addCriteria(Criteria.where(GROUP_NAME).is(groupName)).addCriteria(Criteria.where(COMPLETE_STATUS).is(PROCESSED_STATUS));
            if(!mongoOperations.exists(taskQuery, GroupDataTask.class)){
                log.error("getFilterQuery: group: {} not exists", groupName);
                result.returnWithoutValue(ResultCode.PARAM_ERROR);
                return result;
            }
            query = getDifferentQuery(groupName, bagName, topicName, frameNo, type);
        }
        result.returnWithValue(ResultCode.SUCCESS, query);
        return result;
    }

    private List<GetMetadata> getMetadataListByQuery(Query query, int type) throws IOException {
        List<GetMetadata> metadataList = new ArrayList<>();
        switch (type){
            case GROUP_TYPE:
                List<GroupData> groupDataList = mongoOperations.find(query, GroupData.class);
                for (GroupData groupData : groupDataList) {
                    Map<String, byte[]> metadata = groupData.getMetadata();
                    GetMetadata getMetadata = new GetMetadata();
                    IndexData indexData = new IndexData();
                    indexData.setGroupName(groupData.getGroupName());
                    getMetadata.setIndexData(indexData);
                    List<Pair<String, byte[]>> pairList = traversalMetadata(metadata, groupPathKey);
                    getMetadata.setData(pairList);
                    //metadata.forEach((key, value) -> builder.addMetadatas(Metadata.newBuilder().setKey(key).setValue(ByteString.copyFrom(value)).build()));
                    metadataList.add(getMetadata);
                }
                break;
            case BAG_TYPE:
                List<BagData> bagDataList = mongoOperations.find(query, BagData.class);
                for (BagData bagData : bagDataList) {
                    Map<String, byte[]> metadata = bagData.getMetadata();
                    GetMetadata getMetadata = new GetMetadata();
                    IndexData indexData = new IndexData();
                    indexData.setGroupName(bagData.getGroupName());
                    indexData.setBagName(bagData.getBagName());
                    getMetadata.setIndexData(indexData);
                    List<Pair<String, byte[]>> pairList = traversalMetadata(metadata, bagPathKey);
                    getMetadata.setData(pairList);
                    //metadata.forEach((key, value) -> builder.addMetadatas(Metadata.newBuilder().setKey(key).setValue(ByteString.copyFrom(value)).build()));
                    metadataList.add(getMetadata);
                }
                break;
            case TOPIC_TYPE:
                List<TopicData> topicDataList = mongoOperations.find(query, TopicData.class);
                for (TopicData topicData : topicDataList) {
                    Map<String, byte[]> metadata = topicData.getMetadata();
                    GetMetadata getMetadata = new GetMetadata();
                    IndexData indexData = new IndexData();
                    indexData.setGroupName(topicData.getGroupName());
                    indexData.setBagName(topicData.getBagName());
                    indexData.setTopicName(topicData.getTopicName());
                    getMetadata.setIndexData(indexData);
                    List<Pair<String, byte[]>> pairList = traversalMetadata(metadata, topicPathKey);
                    getMetadata.setData(pairList);
                    //metadata.forEach((key, value) -> builder.addMetadatas(Metadata.newBuilder().setKey(key).setValue(ByteString.copyFrom(value)).build()));
                    metadataList.add(getMetadata);
                }
                break;
            case FRAME_TYPE:
                List<FrameData> frameDataList = mongoOperations.find(query, FrameData.class);
                for (FrameData frameData : frameDataList) {
                    Map<String, byte[]> metadata = frameData.getMetadata();
                    GetMetadata getMetadata = new GetMetadata();
                    IndexData indexData = new IndexData();
                    indexData.setGroupName(frameData.getGroupName());
                    indexData.setBagName(frameData.getBagName());
                    indexData.setTopicName(frameData.getTopicName());
                    indexData.setFrameNo(frameData.getFrameNo());
                    getMetadata.setIndexData(indexData);
                    List<Pair<String, byte[]>> pairList = traversalMetadata(metadata, framePathKey);
                    getMetadata.setData(pairList);
                    //metadata.forEach((key, value) -> builder.addMetadatas(Metadata.newBuilder().setKey(key).setValue(ByteString.copyFrom(value)).build()));
                    metadataList.add(getMetadata);
                }
                break;
             default:
                 throw new TypeNotPresentException("getMetadataListByQuery: type error", new Throwable());
        }
        return metadataList;
    }

    private List<IndexData> getIndexDataByQuery(Query query, int type){
        List<IndexData> indexDataList = new ArrayList<>();
        switch (type){
            case GROUP_TYPE:
                query.fields().include(GROUP_NAME);
                List<GroupData> groupDataList = mongoOperations.find(query, GroupData.class);
                groupDataList.forEach(groupData -> indexDataList.add(new IndexData(groupData.getGroupName(), STRING_DEFAULT, STRING_DEFAULT, INT_DEFAULT)));
                break;
            case BAG_TYPE:
                query.fields().include(BAG_NAME).include(GROUP_NAME);
                List<BagData> bagDataList = mongoOperations.find(query, BagData.class);
                bagDataList.forEach(bagData -> indexDataList.add(new IndexData(bagData.getGroupName(), bagData.getBagName(), STRING_DEFAULT, INT_DEFAULT)));
                break;
            case TOPIC_TYPE:
                query.fields().include(TOPIC_NAME).include(BAG_NAME).include(GROUP_NAME);
                List<TopicData> topicDataList = mongoOperations.find(query, TopicData.class);
                topicDataList.forEach(topicData -> indexDataList.add(new IndexData(topicData.getGroupName(), topicData.getBagName(), topicData.getTopicName(), INT_DEFAULT)));
                break;
            default:
                log.error("getIndexDataByQuery: type error");
        }
        return indexDataList;
    }


    // save


    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result<Long> saveMetadata(@NotNull IndexData indexData, @NotNull Class<?> T, int type, @NotNull Pair<String, byte[]>... metadatas) {
        return saveMetadata(indexData, T, type, new ArrayList<>(Arrays.asList(metadatas)));
    }

    @Retryable(
            value = {MongoException.class, UncategorizedMongoDbException.class},
            maxAttempts = 20,
            backoff = @Backoff(value = 500L, multiplier = 1.5, maxDelay = 15000L)
    )
    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result<Long> saveMetadata(@NotNull IndexData indexData, @NotNull Class<?> T, int type, @NotNull List<Pair<String, byte[]>> metadatasList) {
        Result<Long> result = new Result<>();
        String groupName = indexData.getGroupName();
        String bagName = indexData.getBagName();
        String topicName = indexData.getTopicName();
        int frameNo = indexData.getFrameNo();
        log.info("saveMetadata: group name: {}, bag name: {}, topic name: {}, frame no: {}, metadata size: {}, type: {}", groupName, bagName, topicName, frameNo, metadatasList.size(), type);


//        Result<Query> queryResult = getFilterQuery(groupName, bagName, topicName, frameNo, type);
//        if(queryResult.getCode() != ResultCode.SUCCESS.getCode()){
//            result.returnWithOutValue(queryResult);
//            return result;
//        }
//        Query query = queryResult.getData();
        // 更新 document
        Query query = getDifferentQuery(groupName, bagName, topicName, frameNo, type);
        Update update = new Update();
        metadatasList.forEach(metadata -> update.set(METADATA_POINT+metadata.getKey(), metadata.getValue()));
        update.set(METADATA_UPDATE_TIME, TimeUtil.getNowTimestampLong().toString().getBytes());
        UpdateResult updateResult = mongoOperations.updateMulti(query, update, T);
        result.returnWithValue(ResultCode.SUCCESS, updateResult.getModifiedCount());
        return result;
    }


    // get metadata


    @Override
    public Result<List<GetMetadata>> getMetadataWithoutVerify(@NotNull IndexData indexData, int type, @NotNull String... keys) {
        return getMetadataWithoutVerify(indexData, type, new ArrayList<>(Arrays.asList(keys)));
    }

    @Override
    public Result<List<GetMetadata>> getMetadataWithoutVerify(@NotNull IndexData indexData, int type, @NotNull List<String> keyList) {
        String methodName = new Exception().getStackTrace()[0].getMethodName();
        Result<List<GetMetadata>> result = new Result<>();
        String groupName = indexData.getGroupName();
        String bagName = indexData.getBagName();
        String topicName = indexData.getTopicName();
        int frameNo = indexData.getFrameNo();
        log.info("{}: group name: {}, bag name: {}, topic name: {}, frame no: {}, key size: {}, type: {}", methodName, groupName, bagName, topicName, frameNo, keyList.size(), type);
        Query query = new Query();
        if (!groupName.isEmpty())  query = getDifferentQuery(groupName, bagName, topicName, frameNo, type);
        query.fields().include(GROUP_NAME).include(BAG_NAME).include(FRAME_NO).include(TOPIC_NAME);
        if (keyList.isEmpty()){
            query.fields().include(METADATA);
        }
        else{
            for (String key : keyList) {
                query.fields().include(METADATA_POINT+key);
            }
        }
        try{
            List<GetMetadata> getMetadataList = getMetadataListByQuery(query, type);
            result.returnWithValue(ResultCode.SUCCESS, getMetadataList);
        } catch (IOException e) {
            e.printStackTrace();
            result.returnWithoutValue(ResultCode.DATA_READ_ERROR);
            return result;
        }
        return result;
    }

    @Override
    public Result<List<GetMetadata>> getMetadata(@NotNull IndexData indexData, int type, @NotNull String... keys) {
        return getMetadata(indexData, type, new ArrayList<>(Arrays.asList(keys)));
    }

    @Override
    public Result<List<GetMetadata>> getMetadata(@NotNull IndexData indexData, int type, @NotNull List<String> keyList) {
        Result<List<GetMetadata>> result = new Result<>();
        String groupName = indexData.getGroupName();
        String bagName = indexData.getBagName();
        String topicName = indexData.getTopicName();
        int frameNo = indexData.getFrameNo();
        log.info("getMetadata: group name: {}, bag name: {}, topic name: {}, frame no: {}, key size: {}, type: {}", groupName, bagName, topicName, frameNo, keyList.size(), type);

        Result<Query> queryResult = getFilterQuery(groupName, bagName, topicName, frameNo, type);
        if(queryResult.getCode() != ResultCode.SUCCESS.getCode()){
            result.returnWithOutValue(queryResult);
            return result;
        }
        Query query = queryResult.getData();
        query.with(new Sort(Sort.Direction.ASC, "_id"));
//        Query query = getDifferentQuery(groupName, bagName, topicName, frameNo, type);
        query.fields().include(GROUP_NAME).include(BAG_NAME).include(FRAME_NO).include(TOPIC_NAME);
        if (keyList.isEmpty())
            query.fields().include(METADATA);
        else
            keyList.forEach(key -> query.fields().include(METADATA_POINT+key));

        // 遍历 metadata 把数据读到 list
        try{
            List<GetMetadata> metadataList = getMetadataListByQuery(query, type);
            result.returnWithValue(ResultCode.SUCCESS, metadataList);
        } catch (IOException e) {
            e.printStackTrace();
            result.returnWithoutValue(ResultCode.DATA_READ_ERROR);
        }
        return result;
    }

    @Override
    public Result<List<IndexData>> getIndex(@NotNull IndexData indexData, int type) {
        Result<List<IndexData>>  result = new Result<>();
        String groupName = indexData.getGroupName();
        String bagName = indexData.getBagName();
        String topicName = indexData.getTopicName();
        log.info("getIndex: group name: {}, bag name: {}, topic name: {}, type: {}", groupName, bagName, topicName, type);

        Result<Query> queryResult = getFilterQuery(groupName, bagName, topicName, 0, type);
        if(queryResult.getCode() != ResultCode.SUCCESS.getCode()){
            result.returnWithOutValue(queryResult);
            return result;
        }
        Query query = queryResult.getData();
        query.with(new Sort(Sort.Direction.ASC, "_id"));
        //Query query = getDifferentQuery(groupName, bagName, topicName, 0, type);
        List<IndexData> indexDataList = getIndexDataByQuery(query, type);
        result.returnWithValue(ResultCode.SUCCESS, indexDataList);
        return result;
    }

    @Override
    public Result<List<GetBinData>> getBinData(@NotNull IndexData indexData) {
        String methodName = new Exception().getStackTrace()[0].getMethodName();
        Result<List<GetBinData>> result = new Result<>();
        String groupName = indexData.getGroupName();
        String bagName = indexData.getBagName();
        String topicName = indexData.getTopicName();
        int frameNo = indexData.getFrameNo();
        log.info("{}: group name: {}, bag name: {}, topic name: {}, frame no: {}", methodName, groupName, bagName, topicName, frameNo);

        Result<Query> queryResult = getFilterQuery(groupName, bagName, topicName, frameNo, FRAME_TYPE);
        if(queryResult.getCode() != ResultCode.SUCCESS.getCode()){
            result.returnWithOutValue(queryResult);
            return result;
        }
        Query query = queryResult.getData();
        query.with(new Sort(Sort.Direction.ASC, "_id"));
//        Query query = getDifferentQuery(groupName, bagName, topicName, frameNo, FRAME_TYPE);
        query.fields().include(GROUP_NAME).include(BAG_NAME).include(FRAME_NO).include(TOPIC_NAME).include(BIN_PATH);
        List<FrameData> frameDataList = mongoOperations.find(query, FrameData.class);

        // 读取文件信息并放入 list
        List<GetBinData> binDataList = new ArrayList<>();
        List<Callable<Result>> callableList = new ArrayList<>();
        for (FrameData frameData : frameDataList) {
            IndexData frameIndexData = new IndexData(frameData.getGroupName(), frameData.getBagName(), frameData.getTopicName(), frameData.getFrameNo());
                callableList.add(() -> {
                    Result threadResult = new Result();
                    try {
                        byte[] data = readByteFile(frameData.getBinPath());
                        GetBinData getBinData = new GetBinData(frameIndexData, data);
                        binDataList.add(getBinData);
                        result.returnWithoutValue(ResultCode.SUCCESS);
                    } catch (IOException e) {
                        e.printStackTrace();
                        threadResult.returnWithoutValue(ResultCode.DATA_READ_ERROR);
                        throw e;
                    }
                    return result;
                });
        }

        try {
            ThreadManager.invokeAll(callableList);
        } catch (InterruptedException e) {
            e.printStackTrace();
            result.returnWithoutValue(ResultCode.SYSTEM_ERROR);
            return result;
        }

        result.returnWithValue(ResultCode.SUCCESS, binDataList);
        return result;
    }

}
